package ru.algorithm.join;

import ru.common.Pair;

import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;

import static ru.algorithm.join.DataStreamAlgorithm.read;
import static ru.common.CommonUtils.toList;
import static ru.common.Pair.pair;

/**
 * Yuri Zemlyanskiy <urikz@yandex-team.ru>
 * Date: 05.05.2011
 */
public class SimpleJoinAlgorithm {

    private JoinCondition<double[]> condition;

    public SimpleJoinAlgorithm(final JoinCondition<double[]> joinCondition) {
        this.condition = joinCondition;
    }

    public List<Pair<double[], double[]>> process(InputStream input) {
        final List<double[]> entries = toList(read(input));
        final List<Pair<double[], double[]>> result = new ArrayList<Pair<double[], double[]>>();

        int joined = 0;
        for (int i = 0; i < entries.size(); i++) {
            final double[] first = entries.get(i);
            for (int j = i + 1; j < entries.size(); j++) {
                final double[] second = entries.get(j);
                if (condition.isJoin(first, second)) {
                    result.add(pair(first, second));
                }
            }
        }

        return result;
    }
}
